package com.zh.media.message;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zh.common.constant.AuthConstant;
import com.zh.media.service.ArticleService;
import com.zh.media.service.VideoService;
import com.zh.web.domain.media.entity.Article;
import com.zh.web.domain.media.entity.Video;
import com.zh.web.base.BaseModel;
import com.zh.web.domain.media.vo.OriginalArtVo;
import com.zh.web.enums.Status;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jsoup.Jsoup;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.*;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class MediaConsumer {

    private final ObjectMapper objectMapper;
    private final ArticleService articleService;
    private final VideoService videoService;
    private final CopyOnWriteArrayList<Article> articleList = new CopyOnWriteArrayList<>();
    private final CopyOnWriteArrayList<Video> videoList = new CopyOnWriteArrayList<>();
    private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
    private ScheduledFuture<?> scheduledFuture;
    private Instant lastUpdatedTime = Instant.now();

    @PostConstruct
    private void init() {
        taskScheduler.initialize();
        scheduledFuture = taskScheduler.scheduleWithFixedDelay(this::checkListSize, 3000);
        log.info("初始化文章消费任务调度器");
    }

    @PreDestroy
    private void destroy() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(false);
        }
        taskScheduler.shutdown();
        log.info("销毁文章消费任务调度器");
    }

    @RabbitListener(
            bindings =
            @QueueBinding(
                    value = @Queue("ls.media.article.queue"),
                    exchange = @Exchange(value = "ls.media.exchange"),
                    key = "article"
            )
    )
    public void receiveArticle(Message message) throws IOException {
        try {
            Article article = objectMapper.readValue(message.getBody(), Article.class);
            log.info("接收到 文章 爬虫消息: {}", article);
            this.articleList.add(article);
        } catch (JsonProcessingException e) {
            log.error("解析消息失败", e);
        }
        lastUpdatedTime = Instant.now();
        if (this.articleList.size() >= 10) {
            log.info(" 文章 爬虫消息已满,开始处理...");
            handleResult();
        }
    }

    @RabbitListener(
            bindings =
            @QueueBinding(
                    value = @Queue("ls.media.video.queue"),
                    exchange = @Exchange(value = "ls.media.exchange"),
                    key = "video"
            )
    )
    public void receiveVideo(Message message) throws IOException {
        Video video = objectMapper.readValue(message.getBody(), Video.class);
        log.info("接收到 视频 爬虫消息: {}", video);
        videoList.add(video);
        lastUpdatedTime = Instant.now();
        if (videoList.size() >= 10) {
            log.info(" 视频 爬虫消息已满,开始处理...");
            handleResult();
        }
    }

    private synchronized void handleResult() {
        if (CollUtil.isNotEmpty(this.articleList)) {
            this.handleArticleResult(this.articleList);
        }
        if (CollUtil.isNotEmpty(this.videoList)) {
            this.handleVideoResult(this.videoList);
        }
    }

    private void handleArticleResult(List<Article> list) {
        log.info("开始处理 文章 结果...");
        list.sort(((o1, o2) -> {
            OriginalArtVo originalVo1 = JSON.parseObject(o1.getExtendContent(), OriginalArtVo.class);
            OriginalArtVo originalVo2 = JSON.parseObject(o2.getExtendContent(), OriginalArtVo.class);
            return originalVo2.getCreateTime().compareTo(originalVo1.getCreateTime());
        }));
        OriginalArtVo first = JSON.parseObject(list.get(0).getExtendContent(), OriginalArtVo.class);
        OriginalArtVo last = JSON.parseObject(list.get(list.size() - 1).getExtendContent(), OriginalArtVo.class);
        log.info("处理时间:{}-处理{}条 文章 数据 时间范围:{}-{}", LocalDateTime.now(), list.size(), last.getCreateTime(), first.getCreateTime());
        //保存到数据库
        List<Article> collect = list.stream().peek(article -> {
            String content = article.getContent();
            String text = Jsoup.parse(content).text();
            article.setSummary(StrUtil.sub(text, 0, 201));
            article.setContentType("story");
            article.setCreationType("transshipment");
            article.setLawType("other");
            article.setStatus(Status.REVIEWING);
            article.setCreateOperator(AuthConstant.ADMIN_ID);
            article.setUpdateOperator(AuthConstant.ADMIN_ID);
            article.setCreateTime(LocalDateTime.now());
            article.setUpdateTime(LocalDateTime.now());
        }).collect(Collectors.toList());
        articleService.saveList(collect);
        list.clear();
    }

    private void handleVideoResult(List<Video> list) {
        log.info("开始处理 视频 结果...");
        list.sort(Comparator.comparing(BaseModel::getCreateTime));
        log.info("处理{}条 视频 数据 时间范围:{}-{}", list.size(), list.get(list.size() - 1).getCreateTime(), list.get(0).getCreateTime());
        log.info(" 视频 爬虫消息已保存:{}", list);
        list.clear();
    }

    private void checkListSize() {
        this.checkArticleListSize();
        this.checkVideoListSize();
    }

    private void checkArticleListSize() {
        if (!this.articleList.isEmpty() && Duration.between(lastUpdatedTime, Instant.now()).toSeconds() >= 3) {
            log.info("检测到 文章 爬虫消息超过3s未处理,开始处理...");
            handleResult();
        }
    }

    private void checkVideoListSize() {
        if (!this.videoList.isEmpty() && Duration.between(lastUpdatedTime, Instant.now()).toSeconds() >= 3) {
            log.info("检测到 视频 爬虫消息超过3s未处理,开始处理...");
            handleResult();
        }
    }
}
